package io.kiki.stack.netty;

import io.kiki.stack.netty.channelManager.*;
import io.kiki.stack.netty.command.CommandFactoryImpl;
import io.kiki.stack.netty.config.ClientOption;
import io.kiki.stack.netty.config.GenericOption;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.invoker.*;
import io.kiki.stack.netty.lifeCycle.LifeCycleException;
import io.kiki.stack.netty.userProcessor.UserProcessor;
import io.kiki.stack.netty.userProcessor.UserProcessorRegisterHelper;
import io.kiki.stack.netty.util.AddressParser;
import io.kiki.stack.netty.util.NamedThreadFactory;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


@Slf4j
@Setter
@Getter
public class Client extends AbstractClient {


    private final ConcurrentHashMap<String, UserProcessor<?>> userProcessors;
    private final ChannelEventHandler channelEventHandler;
    private final ChannelEventListener channelEventListener;
    // used in RpcClientAdapter (tr-adapter)
    protected Invoker invoker;
    @Setter
    @Getter
    private ChannelManager channelManager;
    private ReconnectManager reconnectManager;

    private ChannelMonitor channelMonitor;
    private ChannelMonitorStrategy channelMonitorStrategy;
    private ScheduledExecutorService scheduledExecutorService;

    public Client() {
        //this.taskScanner = new TaskScanner();
        this.userProcessors = new ConcurrentHashMap<String, UserProcessor<?>>();
        this.channelEventHandler = new RpcChannelEventHandler(this);
        this.channelEventListener = new ChannelEventListener();
    }


    @Deprecated
    public void init() {
        startup();
    }

    /**
     * Shutdown.
     * <p>
     * Notice:<br>
     * <li>Rpc client can not be used any more after shutdown.
     * <li>If you need, you should destroy it, and instantiate another one.
     */
    @Override
    public void shutdown() {
        super.shutdown();

        this.channelManager.shutdown();
        log.warn("Close all Channels from client side!");
        //this.taskScanner.shutdown();
        log.warn("Rpc client shutdown!");
        if (reconnectManager != null) {
            reconnectManager.shutdown();
        }
        if (channelMonitor != null) {
            channelMonitor.shutdown();
        }
        if (scheduledExecutorService != null) {
            scheduledExecutorService.shutdown();
        }
        for (UserProcessor<?> userProcessor : userProcessors.values()) {
            if (userProcessor.isInited()) {
                userProcessor.shutdown();
            }
        }
    }

    @Override
    public void startup() throws LifeCycleException {
        super.startup();

        for (UserProcessor<?> userProcessor : userProcessors.values()) {
            if (!userProcessor.isInited()) {
                userProcessor.startup();
            }
        }


        ChannelSelectStrategy channelSelectStrategy = option(GenericOption.channel_select_strategy);
        if (channelSelectStrategy == null) {
            channelSelectStrategy = new RandomChannelSelectStrategy(this);
        }
        if (this.channelManager == null) {
            ClientChannelManagerImpl clientChannelManager = new ClientChannelManagerImpl(channelSelectStrategy, new ClientChannelFactory(userProcessors, this), channelEventHandler, channelEventListener);

            clientChannelManager.startup();
            this.channelManager = clientChannelManager;
        }
        this.invoker = new ClientInvoker(new CommandFactoryImpl(), this.channelManager);
        //this.taskScanner.add(this.ChannelManager);
        //this.taskScanner.startup();

        if (isChannelMonitorSwitchOn()) {
            channelMonitor = new ChannelMonitor(this.channelManager, new ScheduledDisconnectChannelMonitorStrategy());
            channelMonitor.startup();
            log.warn("Switch on Channel monitor");
        }
        if (isReconnectSwitchOn()) {
            reconnectManager = new ReconnectManager(this.channelManager);
            reconnectManager.startup();

            channelEventHandler.setReconnectManager(reconnectManager);
            log.warn("Switch on reconnect manager");
        }

        scheduledExecutorService = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RpcTaskScannerThread", true));
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

            @Override
            public void run() {
                try {
                    channelManager.scan();
                } catch (Throwable t) {
                    log.error("Exception caught when scannings.", t);
                }
            }

        }, 10000, 10000, TimeUnit.MILLISECONDS);
    }

    @Override
    public void oneway(final String address, final Object request) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.oneway(address, request, null);
    }

    @Override
    public void oneway(final String address, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.oneway(address, request, requestContext);
    }

    @Override
    public void oneway(final Url url, final Object request) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.oneway(url, request, null);
    }

    @Override
    public void oneway(final Url url, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.oneway(url, request, requestContext);
    }

    @Override
    public void oneway(final Channel channel, final Object request) throws RemotingException {
        ensureStarted();
        this.invoker.oneway(channel, request, null);
    }

    @Override
    public void oneway(final Channel channel, final Object request, final RequestContext requestContext) throws RemotingException {
        ensureStarted();
        this.invoker.oneway(channel, request, requestContext);
    }

    @Override
    public Object invokeSync(final String address, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(address, request, null, timeoutMillis);
    }

    @Override
    public Object invokeSync(final String address, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(address, request, requestContext, timeoutMillis);
    }

    @Override
    public Object invokeSync(final Url url, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invokeSync(url, request, null, timeoutMillis);
    }

    @Override
    public Object invokeSync(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(url, request, requestContext, timeoutMillis);
    }

    @Override
    public Object invokeSync(final Channel channel, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(channel, request, null, timeoutMillis);
    }

    @Override
    public Object invokeSync(final Channel channel, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(channel, request, requestContext, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final String address, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeWithFuture(address, request, null, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final String address, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeWithFuture(address, request, requestContext, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final Url url, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeWithFuture(url, request, null, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeWithFuture(url, request, requestContext, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final Channel channel, final Object request, int timeoutMillis) throws RemotingException {
        ensureStarted();
        return this.invoker.invokeWithFuture(channel, request, null, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(final Channel channel, final Object request, final RequestContext requestContext, int timeoutMillis) throws RemotingException {
        ensureStarted();
        return this.invoker.invokeWithFuture(channel, request, requestContext, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final String address, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.invokeWithCallback(address, request, null, callback, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final String address, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.invokeWithCallback(address, request, requestContext, callback, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final Url url, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.invokeWithCallback(url, request, null, callback, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final Url url, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        this.invoker.invokeWithCallback(url, request, requestContext, callback, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final Channel channel, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException {
        ensureStarted();
        this.invoker.invokeWithCallback(channel, request, null, callback, timeoutMillis);
    }

    @Override
    public void invokeWithCallback(final Channel channel, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException {
        ensureStarted();
        this.invoker.invokeWithCallback(channel, request, requestContext, callback, timeoutMillis);
    }

    @Override
    public void addChannelEventProcessor(ChannelEventType type, ChannelEventProcessor processor) {
        this.channelEventListener.add(type, processor);
    }

    @Override
    public void registerUserProcessor(UserProcessor<?> processor) {
        UserProcessorRegisterHelper.registerUserProcessor(processor, this.userProcessors);
        // startup the processor if it registered after component startup
        if (isInited() && !processor.isInited()) {
            processor.startup();
        }
    }

    @Override
    public Channel createStandaloneChannel(String ip, int port, int connectTimeout) throws RemotingException {
        ensureStarted();
        return this.channelManager.create(ip, port, connectTimeout);
    }

    @Override
    public Channel createStandaloneChannel(String address, int connectTimeout) throws RemotingException {
        ensureStarted();
        return this.channelManager.create(address, connectTimeout);
    }

    @Override
    public void closeStandaloneChannel(Channel channel) {
        ensureStarted();
        if (null != channel) {
            channel.close();
        }
    }

    @Override
    public Channel getChannel(String address, int connectTimeout) throws RemotingException, InterruptedException {
        ensureStarted();
        Url url = AddressParser.parse(address);
        return this.getChannel(url, connectTimeout);
    }

    @Override
    public Channel getChannel(Url url, int connectTimeout) throws RemotingException, InterruptedException {
        ensureStarted();
        url.setConnectTimeout(connectTimeout);
        return this.channelManager.getAndCreateIfAbsent(url);
    }

    @Override
    public Map<String, List<Channel>> getAllManagedChannels() {
        ensureStarted();
        return this.channelManager.getAll();
    }

    @Override
    public boolean checkChannel(String address) {
        return checkChannel(address, false);
    }

    @Override
    public boolean checkChannel(String address, boolean createIfAbsent) {
        return checkChannel(address, createIfAbsent, false);
    }

    @Override
    public boolean checkChannel(String address, boolean createIfAbsent, boolean createAsync) {
        ensureStarted();
        Url url = AddressParser.parse(address);
        Channel conn = this.channelManager.get(url.getId());
        try {
            this.channelManager.check(conn);
        } catch (Exception e) {
            log.warn("check failed. address: {}, Channel: {}", address, conn, e);
            if (createIfAbsent) {
                try {
                    if (createAsync) {
                        this.channelManager.asyncCreateChannel(url);
                        return false;
                    } else {
                        Channel channel = this.channelManager.getAndCreateIfAbsent(url);
                        try {
                            this.channelManager.check(channel);
                            return true;
                        } catch (Exception ex0) {
                            return false;
                        }
                    }
                } catch (Exception ex) {
                    log.warn("check failed and try create Channel for {} also failed.", address, e);
                    return false;
                }
            }
            return false;
        }
        return true;
    }

    /**
     * Close all Channel of a address
     *
     * @param addr address
     */
    public void closeChannel(String addr) {
        ensureStarted();
        Url url = AddressParser.parse(addr);
        if (isReconnectSwitchOn() && reconnectManager != null) {
            reconnectManager.disableReconnect(url);
        }
        this.channelManager.remove(url.getId());
    }

    @Override
    public void closeChannel(Url url) {
        ensureStarted();
        if (isReconnectSwitchOn() && reconnectManager != null) {
            reconnectManager.disableReconnect(url);
        }
        this.channelManager.remove(url.getId());
    }

    @Override
    public void enableConnHeartbeat(String address) {
        ensureStarted();
        Url url = AddressParser.parse(address);
        this.enableConnHeartbeat(url);
    }

    @Override
    public void enableConnHeartbeat(Url url) {
        ensureStarted();
        if (null != url) {
            this.channelManager.enableHeartbeat(this.channelManager.get(url.getId()));
        }
    }

    @Override
    public void disableConnHeartbeat(String address) {
        ensureStarted();
        Url url = AddressParser.parse(address);
        this.disableConnHeartbeat(url);
    }

    @Override
    public void disableConnHeartbeat(Url url) {
        ensureStarted();
        if (null != url) {
            this.channelManager.disableHeartbeat(this.channelManager.get(url.getId()));
        }
    }

    @Override
    @Deprecated
    public void enableReconnectSwitch() {
        option(ClientOption.Channel_reconnect_switch, true);
    }

    @Override
    @Deprecated
    public void disableReconnectSwith() {
        option(ClientOption.Channel_reconnect_switch, false);
    }

    @Override
    @Deprecated
    public boolean isReconnectSwitchOn() {
        return option(ClientOption.Channel_reconnect_switch);
    }

    @Override
    @Deprecated
    public void enableChannelMonitorSwitch() {
        option(ClientOption.Channel_monitor_switch, true);
    }

    @Override
    @Deprecated
    public void disableChannelMonitorSwitch() {
        option(ClientOption.Channel_monitor_switch, false);
    }

    @Override
    @Deprecated
    public boolean isChannelMonitorSwitchOn() {
        return option(ClientOption.Channel_monitor_switch);
    }


}
